package cn.edu.recommender

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import java.lang.Thread.sleep
import java.sql.Date
import java.text.SimpleDateFormat
import scala.collection.mutable.ListBuffer


// 定义样例类
case class Movie(mid: Int, name: String, descri: String, timelong: String, issue: String,
                 shoot: String, language: String, genres: String, actors: String, directors: String)

case class Rating(uid: Int, mid: Int, score: Double, timestamp: Int)

case class MongoConfig(uri: String, db: String)

case class Recommendation(mid: Int, score: Double)

case class GenresRecommendation(genres: String, recs: Seq[Recommendation])


object StatisticsRecommender {

  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_MOVIE_COLLECTION = "Movie"

  //统计的表的名称
  val RATE_MORE_MOVIES = "RateMoreMovies"
  val RATE_MORE_RECENTLY_MOVIES = "RateMoreRecentlyMovies"
  val AVERAGE_MOVIES = "AverageMovies"
  val GENRES_TOP_MOVIES = "GenresTopMovies"

  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27017/recommender",
      "mongo.db" -> "recommender"
    )

    val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    val spark = SparkSession.builder().appName("StatisticsRecommender").master(config("spark.cores")).getOrCreate()

    //加入隐式转换
    import spark.implicits._

    val ratingDS = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating]

    val movieDS = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Movie]

    // 转换为表
    ratingDS.createOrReplaceTempView("ratings")
    movieDS.createOrReplaceTempView("movies")

    //统计所有历史数据中每个电影的评分数
    //数据结构 -> mid,count
    val rateMoreMoviesDF = spark.sql(
      s"""
         |select mid, count(1) as count
         |from ratings
         |group by mid
         |""".stripMargin)

    rateMoreMoviesDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


    //统计以月为单位拟每个电影的评分数
    //数据结构 -> mid,count,time

    //注册一个 UDF 函数，用于将 timestamp 装换成年月格式 1260759144000 => 201605
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")
    spark.udf.register("changeDate", (x: Int) => simpleDateFormat.format(new Date(x * 1000L)))

    val rateMoreRecentlyMovies = spark.sql(
      s"""
         |select mid, count(1) as count, yearmonth
         |from
         |(
         |    select mid, score, changeDate(timestamp) as yearmonth
         |    from ratings
         |) a
         |group by yearmonth, mid
         |""".stripMargin)

    rateMoreRecentlyMovies
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_RECENTLY_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //统计每个电影的平均评分
    //数据结构 -> mid,score
    val averageMoviesDF = spark.sql(
      s"""
         |select mid, avg(score) as score
         |from ratings
         |group by mid
         |""".stripMargin)

    averageMoviesDF.cache()

    averageMoviesDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", AVERAGE_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


    // 统计每种电影类型中评分最高的 10 个电影
    //数据结构 -> genre,movies


    val movieWithScore: DataFrame = movieDS.join(averageMoviesDF, Seq("mid"))

    val genrenTopMovies = movieWithScore.select($"mid", $"score", explode(split(lower($"genres"), "\\|")) as "genre")
      .map(row => (row.getAs[String]("genre"), (row.getAs[Int]("mid"), row.getAs[Double]("score"))))
      .rdd
      .groupByKey()
      .map {
        case (genre, items) => GenresRecommendation(genre, items.toList.sortWith(_._2 > _._2).take(10).map(item => Recommendation(item._1, item._2)))
      }
      .toDS()

    genrenTopMovies
      .write
      .option("uri",mongoConfig.uri)
      .option("collection",GENRES_TOP_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    sleep(600000)
    // 关闭 Spark
    spark.stop()
  }
}
